package com.bigdata.mapreduce.topn;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;

/**
 * 输出的key: 自定义的类型，里面包含了 年、月、日、温度属性
 * 输出的value: 温度
 */
public class TopnMapper extends Mapper<LongWritable, Text, Tkey, IntWritable> {

    private Tkey mkey = new Tkey();
    private IntWritable mval = new IntWritable();

    private Map<Integer, String> cityMap = new HashMap<>();

    @Override
    protected void setup(Mapper<LongWritable, Text, Tkey, IntWritable>.Context context) throws IOException, InterruptedException {
        URI cacheFile = context.getCacheFiles()[0];
        String cachePathStr = cacheFile.getPath();
        System.out.println("cachePath:" + cachePathStr);

        // 缓存路径对应的HDFS Path对象
        Path cachePath = new Path(cachePathStr);
        // HDFS api操作连接对象
        FileSystem fs = cachePath.getFileSystem(context.getConfiguration());
        // 字节流
        FSDataInputStream inputStream = fs.open(cachePath);
        // 字符流
        BufferedReader br = new BufferedReader(new InputStreamReader(inputStream));

        String line;

        while ((line = br.readLine()) != null) {
            String[] cityArr = line.split("\t");
            cityMap.put(Integer.valueOf(cityArr[0]), cityArr[1]);
        }

        System.out.println(cityMap);

    }

    public static void main(String[] args) throws ParseException {
        String str = "2019-6-2 22:22:22\t38";
        Tkey tkey = new TopnMapper().parseLine(str);
        System.out.println(tkey);

    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Tkey, IntWritable>.Context context) throws IOException, InterruptedException {
        /*
        读取一行数据，拆分解析，从字符串中得到年、月、日、温度，
        创建Tkey对象，并且把 年、月、日、温度 设置到 Tkey的对象中
        封装key，value，使用context write
         */
        System.out.println(value.toString());
        try {
            Tkey tkey = parseLine(value.toString());
            // System.out.println("tkey == mkey是否是一个对象？？？" + (tkey == mkey));
            mval.set(tkey.getTemp());
            context.write(tkey, mval);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 解析一行数据，得到Tkey 类型，并且对属性填充值
     *
     * @param str 待解析的字符串
     * @return Tkey对象
     * @throws ParseException
     */
    private Tkey parseLine(String str) throws ParseException {
        // 2019-6-1 22:22:22	36	1
        String[] dataArr = str.split("\t");
        String temp = dataArr[dataArr.length - 2];

        String dateStr = dataArr[0];
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        Date date = sdf.parse(dateStr);

        // 日历工具类
        Calendar calendar = Calendar.getInstance();
        calendar.setTime(date);

        int year = calendar.get(Calendar.YEAR);
        int month = calendar.get(Calendar.MONTH) + 1;
        int day = calendar.get(Calendar.DAY_OF_MONTH);

        mkey.setYear(year);
        mkey.setMonth(month);
        mkey.setDay(day);
        mkey.setTemp(Integer.valueOf(temp));


        // 获取cityId，通过cityId从map中获取到cityName
        Integer cityId = Integer.valueOf(dataArr[dataArr.length - 1]);
        mkey.setCityName(cityMap.get(cityId));

        return mkey;
    }

}
